package com.example.capsule_kafka.kafka;

import com.example.capsule_kafka.bean.ChatRecordBean;
import com.example.capsule_kafka.bean.MessageBean;
import com.example.capsule_kafka.configuration.KafkaConstant;
import com.example.capsule_kafka.impl.ChatRecordServiceImpl;
import com.google.gson.Gson;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @Author: zhuzhiming
 * @Date: 2022/5/17 1:48
 */
@Component
public class EsTypeConsumer {

    @Autowired
    ChatRecordServiceImpl chatRecordService;

    @KafkaListener(topics = KafkaConstant.ES_DOC_INSERT_TOPIC)
    public void insertDoc(String records) {

       MessageBean messageBean = new Gson().fromJson(records, MessageBean.class);
        if (messageBean != null) {
            ChatRecordBean chatRecordBean = new ChatRecordBean();
            chatRecordBean.setField(messageBean.getField());
            chatRecordBean.setRoomName(messageBean.getRoomName());
            chatRecordBean.setSenderId(messageBean.getSenderId());
            chatRecordBean.setSenderName(messageBean.getSenderName());
            chatRecordBean.setContent(messageBean.getContent());
            chatRecordBean.setSendTime(messageBean.getSendTime());
            chatRecordService.insert(chatRecordBean.getField().toLowerCase(),chatRecordBean);
        }

    }

}
